package org.offsync.async

import collection.JavaConversions._
import java.nio.channels.spi.SelectorProvider
import java.nio.channels.SelectionKey
import java.nio.channels.ClosedSelectorException
import org.offsync.rich._
import org.offsync.util.Disposer
import org.offsync.util.DisposableBase
import scala.collection.{mutable => mutable}
import java.nio.channels.Selector
import java.nio.channels.ClosedChannelException

sealed class SpecializedSelector(
    val op: Int) extends DisposableBase {
  val selector = SelectorProvider.provider().openSelector()
  var action: List[() => Unit] = Nil
  
  def invoke(
      f: () => Unit): AsyncReturn = {
    action = List(f)
    selector.wakeup()
    AsyncReturn
  }
  
  val thread = new Thread() {
    override def run() {
      try {
        while (isLive) {
          val selectCount = selector.select()
          
          if (selectCount == 0) {
            val myActions = lock.synchronized {
              val temp = action
              action = Nil
              temp
            }
            
            for (f <- myActions) {
              f()
            }
          } else {
            selector.selectedKeys.iterator.consume { key =>
              assert(try { key.readyOps == op } catch { case e => true });
              key.attachment.asInstanceOf[() => Unit]()
              key.cancel()
            }
          }
        }
      } catch {
        case e: ClosedSelectorException =>
        case e: ClosedChannelException =>
      } finally {
        onThreadExit()
      }
    }
  }
  
  def onThreadExit(): Unit = {}
  
  def start() = thread.start()

  def isOpen(): Boolean = selector.isOpen()
  
  def provider(): SelectorProvider = selector.provider()
  
  def keys(): mutable.Set[SelectionKey] = selector.keys()
  
  def selectedKeys(): mutable.Set[SelectionKey] = selector.selectedKeys()
  
  def selectNow(): Int = selector.selectNow()
  
  def select(timeout: Long): Int = selector.select(timeout)
  
  def select(): Int = selector.select()
  
  def wakeup(): Selector = selector.wakeup()
  
  protected override def onDispose(): Unit = {
    try selector.close() catch { case e => }
    
    super.onDispose()
  }
}

class AcceptSelector extends SpecializedSelector(SelectionKey.OP_ACCEPT)
class ConnectSelector extends SpecializedSelector(SelectionKey.OP_CONNECT)
class ReadSelector extends SpecializedSelector(SelectionKey.OP_READ)
class WriteSelector extends SpecializedSelector(SelectionKey.OP_WRITE)
